/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service;

import com.asiainfo.redis.entity.Cluster;
import com.asiainfo.redis.entity.ClusterNode;
import com.asiainfo.redis.entity.ClusterNodeConfig;
import com.asiainfo.redis.entity.Host;
import com.asiainfo.redis.repository.ClusterNodeConfigRepository;
import com.asiainfo.redis.repository.ClusterNodeRepository;
import com.asiainfo.redis.repository.ClusterRepository;
import com.asiainfo.redis.service.clusterhandler.AddNodes;
import com.asiainfo.redis.service.clusterhandler.AddOldMasterNodesAsSlave;
import com.asiainfo.redis.service.clusterhandler.MigrateSlots;
import com.asiainfo.redis.service.clusterhandler.RedisClusterCallback;
import com.asiainfo.redis.service.clusterhandler.RedisClusterHandler;
import com.asiainfo.redis.service.clusterhandler.RemoveNode;
import com.asiainfo.redis.service.clusterhandler.RemoveSlaveNodes;
import com.asiainfo.redis.service.redis.MyJedis;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.jedis.JedisClusterConnection;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Protocol;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.HashMap;
import java.util.TreeMap;
import java.util.Collection;
import java.util.Set;

/**
 * @version v 1.0 on 2016/3/2 16:19
 * @auther william.xu
 */
@Service
public class RedisService {

    protected static final Logger LOGGER = LoggerFactory.getLogger(RedisService.class);

    @Resource
    private HostService hostService;

    @Resource
    private ClusterRepository clusterRepository;

    @Resource
    private ClusterNodeRepository clusterNodeRepository;

    @Resource
    private ClusterNodeConfigRepository clusterNodeConfigRepository;

    //@Resource
    //private RedisClusterHandler redisClusterHandler;

    public final static String CHARSET = "UTF-8";

    public final static Integer SLEEP_TIME = 5 * 1000;
    private final static Integer MBYTES = 1024 * 1024;
    public final static Integer CLUSTER_MIN_NUM = 3;


    public List<Cluster> getClusterList(boolean isMasterClusterOnly) {
        if (isMasterClusterOnly){
            return clusterRepository.getMasterClusterList();
        }else{
            return clusterRepository.findAll();
        }
    }

    public Long getStandbyClusterId(Long clusterId){
        Cluster cluster = clusterRepository.findOne(clusterId);
        Assert.notNull(cluster, "该集群不存在，clusterId="+clusterId);
        return cluster.getStandbyId();
    }

    /**
     * 新建集群
     *
     * @param cluster
     */
    @Transactional(rollbackFor = Exception.class)
    public void newCluster(final Cluster cluster) throws Exception {

        final List<ClusterNode> masterNodes = cluster.getMasterNodes();

        //1、保存到数据库
        //1.1保存redis_cluster表
        clusterRepository.save(cluster);
        //1.2保存redis_cluster_node表
        //查询集群节点时，走spring redis的api获取具体节点信息
        for (ClusterNode masterNode : masterNodes) {
            Host host = hostService.findHost(masterNode.getHostId());
            masterNode.setIp(host.getIp());
            masterNode.setClusterId(cluster.getId());
            clusterNodeRepository.save(masterNode);
            List<ClusterNode> slaveNodes = masterNode.getSlaveNodes();
            for (ClusterNode slaveNode : slaveNodes) {
                host = hostService.findHost(slaveNode.getHostId());
                slaveNode.setIp(host.getIp());
                slaveNode.setClusterId(cluster.getId());
                clusterNodeRepository.save(slaveNode);
            }
        }

        //1.3保存REDIS_CLUSTER_NODE_CONFIG表
        ClusterNodeConfig clusterNodeConfig = cluster.getClusterNodeConfig();
        clusterNodeConfig.setClusterId(cluster.getId());
        clusterNodeConfigRepository.save(clusterNodeConfig);

        //2、根据redisClusterNodeConfig，ssh到目标host（遍历所有的masternode和slavenode）上执行生成集群节点的初始化脚本，
        // 并启动对应的node
        for (ClusterNode masterNode : masterNodes) {
            hostService.initRedisNode(clusterNodeConfig, masterNode);
            hostService.startRedisNode(masterNode);
            List<ClusterNode> slaveNodes = masterNode.getSlaveNodes();
            for (ClusterNode slaveNode : slaveNodes) {
                hostService.initRedisNode(clusterNodeConfig, slaveNode);
                hostService.startRedisNode(slaveNode);
            }
        }
        Thread.sleep(SLEEP_TIME);
        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(cluster.getId());
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                //3、redis-cli到其中一个节点，进行meet操作
                redisClusterHandler.meet();
                //一个空集群只要meet了一个node后，必须还需要等待100秒(JedisClusterConnection的getTopology方法会缓存100ms刷新node信息),
                // 才可以使用spring的RedisClusterConnection对象进行操作，它封装了jredis的api，简化了操作
                //否则抛出"Could not retrieve cluster information"的异常
                //等待时间太短，会抛出ClusterStateFailureException: Could not find node at 192.168.198.128:7001. Is your
                // cluster info up to date?
                //meet操作后，需要等待集群join完成
                redisClusterHandler.checkAllMeeting(connection);
                //4、分配slave节点
                redisClusterHandler.replicate(masterNodes, connection);
                //5、给master node平均分配slot
                redisClusterHandler.addSlots(masterNodes, connection);
                return null;
            }
        });

    }

    /**
     * 添加节点到集群
     */
    @Transactional(rollbackFor = Exception.class)
    public void addNodesToCluster(Cluster cluster) throws Exception {

        final List<ClusterNode> masterNodes = cluster.getMasterNodes();

        //RedisCluster rc = redisClusterRepository.findOne(redisCluster.getId());
        final ClusterNodeConfig redisClusterNodeConfig = clusterNodeConfigRepository.findByClusterId(cluster.getId());
        final ClusterNode existNode = clusterNodeRepository.findByClusterId(cluster.getId()).get(0);
        Assert.notNull(existNode, "该集群不存在任何节点");

        //1、保存到数据库
        String clusterName = cluster.getClusterName();
        //1.2保存到redis_cluster_node表
        for (ClusterNode masterNode : masterNodes) {
            Host host = hostService.findHost(masterNode.getHostId());
            List<ClusterNode> list = clusterNodeRepository.findByClusterIdAndHostIdAndPort(cluster.getId(), host
                    .getId(), masterNode.getPort());
            if (list.size() == 0) {
                masterNode.setIp(host.getIp());
                masterNode.setClusterId(cluster.getId());
                clusterNodeRepository.save(masterNode);
            }
            List<ClusterNode> slaveNodes = masterNode.getSlaveNodes();
            for (ClusterNode slaveNode : slaveNodes) {
                host = hostService.findHost(slaveNode.getHostId());
                list = clusterNodeRepository.findByClusterIdAndHostIdAndPort(cluster.getId(), host.getId(), slaveNode
                        .getPort());
                if (list.size() == 0) {
                    slaveNode.setIp(host.getIp());
                    slaveNode.setClusterId(cluster.getId());
                    clusterNodeRepository.save(slaveNode);
                }
            }
        }

        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(cluster.getId());
        redisClusterHandler.execute(new RedisClusterCallback<Object>() {
            @Override
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return new AddNodes(existNode, masterNodes, redisClusterNodeConfig,
                        redisClusterHandler, hostService).doInRedisCluster(connection);
            }
        });

    }

    /**
     * 从集群中删除节点
     *
     * @param nodeId
     * @throws Exception
     */
    @Transactional(rollbackFor = Exception.class)
    public void removeNodeFromCluster(Long nodeId) throws Exception {
        final ClusterNode clusterNode = clusterNodeRepository.findOne(nodeId);
        Assert.notNull(clusterNode, "找不到该节点");

        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterNode.getClusterId());
        redisClusterHandler.execute(new RedisClusterCallback<Object>() {
            @Override
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return new RemoveNode(redisClusterHandler, hostService,
                        clusterNodeRepository, clusterNode).doInRedisCluster(connection);
            }
        });

    }

    /**
     * 迁移节点
     *
     * @param clusterId
     * @param fromNodeId
     * @param toNodeId
     * @param slotsNum
     * @throws Exception
     */
    public void migrateSlots(Long clusterId, final Long fromNodeId, final Long toNodeId, final Integer slotsNum)
            throws Exception {

        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return new MigrateSlots(fromNodeId, toNodeId, slotsNum, redisClusterHandler, clusterNodeRepository).doInRedisCluster
                        (connection);
            }
        });

    }

    public Cluster getClusterAndNodesInfo(final Long clusterId) throws Exception {
        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        return redisClusterHandler.execute(new RedisClusterCallback<Cluster>() {
            public Cluster doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return redisClusterHandler.getClusterAndNodesInfo(clusterId, connection);
            }
        });
    }

    public Properties getInfo(Long clusterNodeId) throws Exception {
        final ClusterNode clusterNode = clusterNodeRepository.findOne(clusterNodeId);
        Assert.notNull(clusterNode, "找不到该节点");
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterNode.getClusterId());
        return redisClusterHandler.execute(new RedisClusterCallback<Properties>() {
            public Properties doInRedisCluster(RedisClusterConnection connection) throws Exception {
                RedisClusterNode node = new RedisClusterNode(clusterNode.getIp(), clusterNode.getPort());
                return connection.info(node);
            }
        });
    }

    public Map getInfo(Long clusterId, final String properties) throws Exception {
        RedisClusterConnection redisClusterConnection = null;
        final Map data = new HashMap();
        final List<ClusterNode> clusterNodeList = clusterNodeRepository.findByClusterId(clusterId);
        for (String property : properties.split(",")) {
            Map result = new TreeMap();
            for (ClusterNode node : clusterNodeList) {
                result.put(node.getIp() + ":" + node.getPort(), "0");
            }
            data.put(property, result);
        }
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                for (ClusterNode node : clusterNodeList) {
                    try {
                        Properties info = connection.info(new RedisClusterNode(node.getIp(), node.getPort()));
                        for (String property : properties.split(",")) {
                            Map result = (Map) data.get(property);
                            parsePropertyData(node, info, result, property);
                        }
                    } catch (Exception e) {
                        LOGGER.warn(e.toString());
                    }
                }
                return null;
            }
        });
        return data;

    }

    private void parsePropertyData(ClusterNode node, Properties info, Map result, String property) {
        Iterator it = info.keySet().iterator();
        while (it.hasNext()) {
            String key = (String) it.next();
            if (key.equals(property)) {
                String value = info.getProperty(key);
                result.put(node.getIp() + ":" + node.getPort(), value);
                if ("used_memory".equals(property)) {
                    result.put(node.getIp() + ":" + node.getPort(), Long.parseLong(value) / (float) (MBYTES));
                }
            }
        }
    }

    public Map getConnectedClients(Long clusterId) throws Exception {
        return getInfo(clusterId, "connected_clients");
    }

    public Map getUsedMemory(Long clusterId) throws Exception {
        return getInfo(clusterId, "used_memory");
    }

    public Map getUsedMemoryRss(Long clusterId) throws Exception {
        return getInfo(clusterId, "used_memory_rss");
    }

    public Map getUsedCpuSys(Long clusterId) throws Exception {
        return getInfo(clusterId, "used_cpu_sys");
    }

    public Map getUsedCpuUser(Long clusterId) throws Exception {
        return getInfo(clusterId, "used_cpu_user");
    }

    public List<String> getKeysBySlot(final Integer slot, Long clusterId) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        return redisClusterHandler.execute(new RedisClusterCallback<List<String>>() {
            public List<String> doInRedisCluster(RedisClusterConnection connection) throws Exception {
                long countKeys = connection.clusterCountKeysInSlot(slot);
                List<byte[]> keyList = connection.clusterGetKeysInSlot(slot, (int) countKeys);
                List<String> list = new ArrayList<String>();
                for (byte[] key : keyList) {
                    list.add(new String(key, CHARSET));
                }
                return list;
            }
        });
    }

    public List<Map> getkeysByKeyQuery(final String keyQuery, Long clusterId) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        return redisClusterHandler.execute(new RedisClusterCallback<List<Map>>() {
            public List<Map> doInRedisCluster(RedisClusterConnection connection) throws Exception {
                Set<byte[]> keyList = connection.keys((keyQuery + "*").getBytes(CHARSET));
                List<Map> list = new ArrayList<Map>();
                for (byte[] key : keyList) {
                    Map map = new HashMap();
                    String keyStr = new String(key, CHARSET);
                    map.put("key", keyStr);
                    //map.put("memory", getMemory(connection, keyStr));//持续遍历调用debug指令，一段时间后线程会永远卡住了
                    map.put("memory", 0);
                    list.add(map);
                }
                return list;
            }
        });
    }

    private long getMemory(RedisClusterConnection connection, String key){
        long memory = 0l;
        try {
            System.out.println(key);
            String debugInfo = ((JedisClusterConnection)connection).debug(key);
            System.out.println(debugInfo);
            String[] info = debugInfo.split(" ");
            for (int i = 0; i < info.length; i++){
                String section = info[i];
                if (section.indexOf("serializedlength") != -1){
                    memory = key.length() + Long.parseLong(section.split(":")[1]);
                    //key_size = strlen(key) + serializedlength + 7
                    break;
                }
            }
        }catch (Exception e){
            LOGGER.error(e.toString());
        }
        return memory;
    }

    public Map getValueByKey(final String key, Long clusterId) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        return redisClusterHandler.execute(new RedisClusterCallback<Map>() {
            public Map doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return getValueByKey(connection, key);
            }
        });
    }

    private Map getValueByKey(RedisClusterConnection connection, String key) throws Exception{
        byte[] keyBytes = key.getBytes(CHARSET);
        Map result = new HashMap();
        DataType dataType = connection.type(keyBytes);
        result.put("dataType", dataType.code());
        if (dataType == DataType.STRING) {
            byte[] value = connection.get(keyBytes);
            if (value != null) {
                result.put("data", new String(value, CHARSET));
            } else {
                result.put("data", null);
            }
        }else if (dataType == DataType.LIST){
            List<byte[]> list = connection.lRange(keyBytes, 0, -1);
            List<Map> valueList = new ArrayList<Map>();
            for(byte[] value : list){
                Map map = new HashMap();
                map.put("value", new String(value, CHARSET));
                valueList.add(map);
            }
            result.put("data", valueList);
        }else if (dataType == DataType.SET){
            Set<byte[]> set = connection.sMembers(keyBytes);
            List<Map> valueList = new ArrayList<Map>();
            for(byte[] value : set){
                Map map = new HashMap();
                map.put("value", new String(value, CHARSET));
                valueList.add(map);
            }
            result.put("data", valueList);
        }else if (dataType == DataType.ZSET){
            //Set<byte[]> zset = connection.zRange(keyBytes, 0, -1);
            Set<RedisZSetCommands.Tuple> zset = connection.zRangeWithScores(keyBytes, 0, -1);
            List<Map> valueList = new ArrayList<Map>();
            for(RedisZSetCommands.Tuple value : zset){
                Map map = new HashMap();
                map.put("value", new String(value.getValue(), CHARSET));
                map.put("score", value.getScore());
                valueList.add(map);
            }
            result.put("data", valueList);
        }else if (dataType == DataType.HASH){
            Map<byte[], byte[]> fieldValues = connection.hGetAll(keyBytes);
            List<Map> valueList = new ArrayList<Map>();
            Set<byte[]> keySet = fieldValues.keySet();
            for (byte[] mykey : keySet){
                Map map = new HashMap();
                map.put("field", new String(mykey, CHARSET));
                map.put("value", new String(fieldValues.get(mykey), CHARSET));
                valueList.add(map);
            }
            result.put("data", valueList);
        }
        return result;

    }

    //value数据格式：
    //{dataType:'string', data:'value'}
    //{dataType:'list', data:[{value:'value1'},{value:'value2'}]}, set和list一致
    //{dataType:'zset', data:[{value:'value1', score: 1.5}, {value:'value2', score: 2.5}]}
    //{dataType:'hash', data:[{field:'field1', value:'value1'}, {field:'field2', value:'value2'}]}
    public void setValueByKey(final String key, final Map value, Long clusterId) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                setValue(connection, key, value);
                return null;
            }
        });
    }

    private void setValue(RedisClusterConnection connection, String key, Map value) throws Exception{
        String dataType = (String)value.get("dataType");
        byte[] keyBytes = key.getBytes(CHARSET);
        connection.del(keyBytes);
        if ("string".equals(dataType)){
            connection.set(keyBytes, ((String)value.get("data")).getBytes(CHARSET));
        }else if ("list".equals(dataType)){
            List<Map> list = (List)value.get("data");
            for (Map item : list){
                connection.rPush(keyBytes, ((String)item.get("value")).getBytes(CHARSET));
            }
        }else if ("set".equals(dataType)){
            List<Map> list = (List)value.get("data");
            for (Map item : list){
                connection.sAdd(keyBytes, ((String)item.get("value")).getBytes(CHARSET));
            }
        }else if ("zset".equals(dataType)){
            List<Map> list = (List)value.get("data");
            for (Map item : list){
                connection.zAdd(keyBytes,
                        Double.parseDouble((String)item.get("score")),
                        ((String)item.get("value")).getBytes(CHARSET));
            }
        }else if ("hash".equals(dataType)){
            List<Map> list = (List)value.get("data");
            for (Map item : list){
                connection.hSet(keyBytes,
                        ((String)item.get("field")).getBytes(CHARSET),
                        ((String)item.get("value")).getBytes(CHARSET));
            }
        }
    }

    public List<Cluster> getStandbyClusterList(Long clusterId){
        return clusterRepository.getStandbyClusterList(clusterId);
    }

    @Transactional(rollbackFor = Exception.class)
    public void setStandbyCluster(Long clusterId, Long standbyId){
        Cluster cluster = clusterRepository.findOne(clusterId);
        Assert.notNull(cluster, "找不到该集群");
        Long oldStandbyId = cluster.getStandbyId();
        //如果原先设置了备用集群，把原来的备用集群改成主用集群
        if (oldStandbyId != null){
            Cluster oldStandby = clusterRepository.findOne(oldStandbyId);
            if (oldStandby != null){
                oldStandby.setType(0);
                oldStandby.setStandbyId(null);
                clusterRepository.save(oldStandby);
            }
        }
        //把standbyId设置成备用状态
        Cluster standby = clusterRepository.findOne(standbyId);
        Assert.notNull(cluster, "找不到该备用集群集群");
        standby.setType(1);
        standby.setStandbyId(null);
        clusterRepository.save(standby);

        cluster.setType(0);
        cluster.setStandbyId(standbyId);
        clusterRepository.save(cluster);

    }

    public void refreshAllData(Long clusterId) throws Exception {

        List<RedisClusterNode> slaveList = new ArrayList<RedisClusterNode>();
        List<RedisClusterNode> masterList = new ArrayList<RedisClusterNode>();
        Map<RedisClusterNode, Collection<RedisClusterNode>> masterSlaveMap = null;

        System.out.println("在cluster中初始化数据");
        initDataOnCluster(clusterId);
        Thread.sleep(SLEEP_TIME);

        System.out.println("移走slave节点");
        masterSlaveMap = removeSlaveNodes(clusterId, slaveList, masterList);
        Thread.sleep(SLEEP_TIME);

        System.out.println("使用slave节点创建新cluster");
        //如果之前就已经初始化jedis cluster api或spring data redis api时，不会出现如下状况：
        //如果此时使用jedis cluster api或spring data redis api时取数据，可能会取到老数据，也可能提示CLUSTERDOWN，说明如下：
        //由于这里启动了独立的slave，当前端lookup到这个slave节点时，会提示CLUSTERDOWN The cluster is down；
        // lookup到原来的master节点时，能取到老的数据
        //以上过程一直持续到slave节点meet完成，新的cluster已经建好，能连到新的集群，取数据正常，为null
        createNewClusterFromSlaveNodes(clusterId, slaveList);
        Thread.sleep(SLEEP_TIME);

        System.out.println("在新cluster中写入新数据");
        refreshDataOnNewCluster(slaveList);
        Thread.sleep(SLEEP_TIME);

        System.out.println("停止原来的master节点");
        //使用jedis cluster api或spring data redis api时，之前一直取到老cluster的数据，不会出现错误。
        // 此时由于开始停止了master节点，client会重试到新的cluster，会过几秒后会取到新的数据，就是这时请求应答比较慢而已，不会出现错误。
        shutdownOldMasterNodes(clusterId, masterList);
        Thread.sleep(SLEEP_TIME);
        System.out.println("把原来的master节点作为slave节点添加到新的cluster中");
        addOldMasterNodesAsSlaveToNewCluster(clusterId, masterList, slaveList, masterSlaveMap);

    }

    private void initDataOnCluster(Long clusterId) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                //测试数据
                connection.set("hello".getBytes(), "world".getBytes());
                connection.set("hello2".getBytes(), "world".getBytes());
                return null;
            }
        });
    }

    private Map<RedisClusterNode, Collection<RedisClusterNode>> removeSlaveNodes(final Long clusterId, final
    List<RedisClusterNode> slaveList, final List<RedisClusterNode> masterList) throws Exception {
        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(clusterId);
        return redisClusterHandler.execute(new RedisClusterCallback<Map<RedisClusterNode,
                Collection<RedisClusterNode>>>() {
            public Map<RedisClusterNode, Collection<RedisClusterNode>> doInRedisCluster(RedisClusterConnection
                                                                                                connection) throws
                    Exception {
                return new RemoveSlaveNodes(redisClusterHandler, hostService, clusterNodeRepository, clusterId,
                        slaveList, masterList).doInRedisCluster(connection);
            }
        });
    }


    private void createNewClusterFromSlaveNodes(Long clusterId, final List<RedisClusterNode> slaveList) throws
            Exception {

        for (RedisClusterNode clusterNode : slaveList) {
            List<ClusterNode> list = clusterNodeRepository.findByClusterIdAndIpAndPort(clusterId, clusterNode.getHost
                    (), clusterNode.getPort());
            if (list.size() > 0) {
                hostService.startRedisNode(list.get(0));
            }
        }
        Thread.sleep(SLEEP_TIME);
        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(slaveList);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                redisClusterHandler.meet();
                redisClusterHandler.checkAllMeeting(connection);
                List<ClusterNode> masterNodes = new ArrayList<ClusterNode>();
                for (RedisClusterNode node : slaveList) {
                    ClusterNode masterNode = new ClusterNode();
                    masterNode.setIp(node.getHost());
                    masterNode.setPort(node.getPort());
                    masterNodes.add(masterNode);
                }
                redisClusterHandler.addSlots(masterNodes, connection);
                return null;
            }
        });

    }

    private void refreshDataOnNewCluster(List<RedisClusterNode> slaveList) throws Exception {
        RedisClusterHandler redisClusterHandler = new RedisClusterHandler(slaveList);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                //测试数据
                connection.set("hello".getBytes(), "world22".getBytes());
                connection.set("hello2".getBytes(), "world22".getBytes());
                return null;
            }
        });
    }

    private void shutdownOldMasterNodes(Long clusterId, List<RedisClusterNode> masterList) throws Exception {
        for (RedisClusterNode node : masterList) {
            //shutdown节点
            Jedis jedis = new Jedis(node.getHost(), node.getPort(), Protocol.DEFAULT_TIMEOUT);
            jedis.shutdown();
            jedis.close();
            //删除nodes.conf文件
            List<ClusterNode> list = clusterNodeRepository.findByClusterIdAndIpAndPort(clusterId, node.getHost(),
                    node.getPort());
            if (list.size() > 0) {
                hostService.deleteNodesConfigFile(list.get(0));
            }
        }
    }

    private void addOldMasterNodesAsSlaveToNewCluster(Long clusterId, final List<RedisClusterNode> masterList, final
    List<RedisClusterNode> slaveList, final Map<RedisClusterNode, Collection<RedisClusterNode>> masterSlaveMap)
            throws Exception {

        //启动原来的master节点
        for (RedisClusterNode master : masterList) {
            List<ClusterNode> list = clusterNodeRepository.findByClusterIdAndIpAndPort(clusterId, master.getHost(),
                    master.getPort());
            if (list.size() > 0) {
                hostService.startRedisNode(list.get(0));
            }
        }
        Thread.sleep(SLEEP_TIME);
        List<RedisClusterNode> nodeList = new ArrayList<>();
        nodeList.addAll(slaveList);
        nodeList.addAll(masterList);
        final RedisClusterHandler redisClusterHandler = new RedisClusterHandler(nodeList);
        redisClusterHandler.execute(new RedisClusterCallback() {
            public Object doInRedisCluster(RedisClusterConnection connection) throws Exception {
                return new AddOldMasterNodesAsSlave(redisClusterHandler, masterList, slaveList, masterSlaveMap);
            }
        });

    }

    public boolean checkRedisClusterStatus(Long clusterId) {

        boolean result = false;
        try {
            Thread.sleep(5000);
        }catch (Exception e){
        }
        int retryCount = 10;
        while(retryCount > 0){

            RedisClusterConnection redisClusterConnection = null;
            try {
                LOGGER.warn("开始尝试检查集群状态, clusterId="+ clusterId);
                redisClusterConnection = RedisClusterHandler.getRedisClusterConnection(clusterId);
                redisClusterConnection.clusterGetNodes();
                result = true;
                break;
            }catch (Exception e){
                LOGGER.warn("检查集群状态出现错误："+e);
            }finally {
                if (redisClusterConnection != null) {
                    RedisClusterHandler.closeRedisClusterConnection(redisClusterConnection);
                }
            }
            try {
                Thread.sleep(5000);
            }catch (Exception e){
            }
            retryCount--;
        }
        return result;

    }

    public void promoteMaster(Long nodeId, boolean isForce){
        ClusterNode clusterNode = clusterNodeRepository.findOne(nodeId);
        Assert.notNull(clusterNode, "找不到该节点");

        MyJedis jedis = null;
        try {
            jedis = new MyJedis(clusterNode.getIp(), clusterNode.getPort(), Protocol.DEFAULT_TIMEOUT);
            if (isForce){
                jedis.clusterFailoverForce();
            }else{
                jedis.clusterFailover();
            }
        }finally {
            if (jedis != null){
                jedis.close();
            }
        }
    }

    public void replicaMigration(Long id, String nodeId){
        ClusterNode clusterNode = clusterNodeRepository.findOne(id);
        Assert.notNull(clusterNode, "找不到该节点");

        Jedis jedis = null;
        try {
            jedis = new Jedis(clusterNode.getIp(), clusterNode.getPort(), Protocol.DEFAULT_TIMEOUT);
            jedis.clusterReplicate(nodeId);
        }finally {
            if (jedis != null){
                jedis.close();
            }
        }
    }


    public void configSet(String param, String value, Long nodeId){
        ClusterNode clusterNode = clusterNodeRepository.findOne(nodeId);
        Assert.notNull(clusterNode, "找不到该节点");

        MyJedis jedis = null;
        try {
            jedis = new MyJedis(clusterNode.getIp(), clusterNode.getPort(), Protocol.DEFAULT_TIMEOUT);
            jedis.configSet(param, value);
            jedis.configRewrite();
            //lua脚本不支持config rewrite指令，所以放弃
            //List<String> keys = new ArrayList<>();
            //List<String> args = new ArrayList<>();
            //jr.eval("redis.call('set', 'SET', KEYS[1], ARGV[1])", keys, args);

        }finally {
            if (jedis != null){
                jedis.close();
            }
        }

    }

}
